package net.cassandraview;

import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.faces.application.FacesMessage;
import javax.faces.bean.ManagedBean;
import javax.faces.bean.ManagedProperty;
import javax.faces.bean.ViewScoped;
import javax.faces.context.FacesContext;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;

import com.google.common.base.Charsets;

@ManagedBean(name = "queryPage")
@ViewScoped
public class QueryPage implements Serializable {

	private final transient Logger log = Logger.getLogger(QueryPage.class.getName());
	
	private static final long serialVersionUID = 1L;
	
	private String keyspace;
	
	private String columnFamily;
	
	private Integer limit = 10;
	
	private String query;
	
	private List<ColumnQueryResult> lastColumnQueryResults;
	
	private List<SuperColumnQueryResult> lastSuperColumnQueryResults;
	
	@ManagedProperty("#{userSession}")
	private UserSession session;
	
	public UserSession getSession() {
		return session;
	}

	public void setSession(UserSession session) {
		this.session = session;
	}

	public String getColumnFamily() {
		return session.getCurrentColumnFamily();
	}
	
	public Boolean isSuperColumn() {
		Cassandra.Client cl = session.getClient();
		List<CfDef> cfs;
		try {
			cfs = cl.describe_keyspace(getKeyspace()).getCf_defs();
			for (CfDef cf : cfs) {
				if (cf.getName().equals(getColumnFamily())) {
					return cf.getColumn_type().equals("Super");
				}
			}
		} catch (Exception e) {
			logFailure("Exception while trying to retrieve data about column family.", e);
		}
		return null;
	}
	
	public List<ColumnQueryResult> getColumns() {
		if (!getKeyspace().equals(keyspace) || !getColFamily().equals(columnFamily) || lastColumnQueryResults == null) {
			lastColumnQueryResults = queryColumns();
			columnFamily = getColFamily();
		}
		return lastColumnQueryResults;
	}
	
	protected List<ColumnQueryResult> queryColumns() {
		Cassandra.Client client = session.getClient();
		List<ColumnQueryResult> result = null;
		try {
			client.set_keyspace(getKeyspace());
            SlicePredicate predicate = new SlicePredicate();
            SliceRange sliceRange = new SliceRange();
            sliceRange.setStart(new byte[0]).setFinish(new byte[0]);
            sliceRange.setCount(Integer.MAX_VALUE);
            predicate.setSlice_range(sliceRange);

			KeyRange range = new KeyRange(limit);
			ByteBuffer startKey = ByteBuffer.wrap(new byte[0]);
			ByteBuffer endKey = ByteBuffer.wrap(new byte[0]);
			range.setStart_key(startKey).setEnd_key(endKey);

			ColumnParent columnParent = new ColumnParent(getColFamily());
			List<KeySlice> keySlices = client.get_range_slices(columnParent, predicate, range, ConsistencyLevel.ONE);
			result = new ArrayList<ColumnQueryResult>(keySlices.size());
			for (KeySlice slice : keySlices) {
				ColumnQueryResult res = new ColumnQueryResult(parseBuffer(slice.bufferForKey()), new ArrayList<Column>());
				for (ColumnOrSuperColumn col : slice.getColumns()) {
					org.apache.cassandra.thrift.Column value = col.getColumn();
					res.getColumns().add(new Column(parseBuffer(value.name), parseBuffer(value.value), new Date(value.timestamp), value.ttl));
				}
				result.add(res);
			}
		} catch (Exception e) {
			logFailure("Exception while trying to retrieve data from db.", e);
		}

		return result;
	}
	
	public List<SuperColumnQueryResult> getSuperColumns() {
		if (!getKeyspace().equals(keyspace) || !getColFamily().equals(columnFamily) || lastSuperColumnQueryResults == null) {
			lastSuperColumnQueryResults = querySuperColumns();
			keyspace = getKeyspace();
			columnFamily = getColFamily();
		}
		return lastSuperColumnQueryResults;
	}
	
	protected List<SuperColumnQueryResult> querySuperColumns() {
		return null;
	}
	
	public Integer getLimit() {
		return limit;
	}

	public void setLimit(Integer limit) {
		this.limit = limit;
	}

	protected String getKeyspace() {
		return session.getCurrentKeyspace() != null ? session.getCurrentKeyspace() : "";
	}
	
	protected String getColFamily() {
		return session.getCurrentColumnFamily()!= null ? session.getCurrentColumnFamily() : "";
	}
	
	/**
	 * Model for results of query to super column 
	 */
	public static class SuperColumnQueryResult {
		private String key;
		private List<ColumnQueryResult> columns;
		public SuperColumnQueryResult(String key, List<ColumnQueryResult> columns) {
			super();
			this.key = key;
			this.columns = columns;
		}
		public String getKey() {
			return key;
		}
		public void setKey(String key) {
			this.key = key;
		}
		public List<ColumnQueryResult> getColumns() {
			return columns;
		}
		public void setColumns(List<ColumnQueryResult> columns) {
			this.columns = columns;
		}
	}
	
	/**
	 * Model for results of query to column 
	 */
	public static class ColumnQueryResult {
		private String key;
		private List<Column> columns;
		public ColumnQueryResult(String key, List<Column> columns) {
			super();
			this.key = key;
			this.columns = columns;
		}
		public String getKey() {
			return key;
		}
		public void setKey(String key) {
			this.key = key;
		}
		public List<Column> getColumns() {
			return columns;
		}
		public void setColumns(List<Column> columns) {
			this.columns = columns;
		}
	}

	/**
	 * Parsed and converted to String representation of Column 
	 */
	public static class Column {
		private String name;
		private String value;
		private Date timestamp;
		private int ttl;
		public Column(String name, String value, Date timestamp, int ttl) {
			super();
			this.name = name;
			this.value = value;
			this.timestamp = timestamp;
			this.ttl = ttl;
		}
		public String getName() {
			return name;
		}
		public void setName(String name) {
			this.name = name;
		}
		public String getValue() {
			return value;
		}
		public void setValue(String value) {
			this.value = value;
		}
		public Date getTimestamp() {
			return timestamp;
		}
		public void setTimestamp(Date timestamp) {
			this.timestamp = timestamp;
		}
		public int getTtl() {
			return ttl;
		}
		public void setTtl(int ttl) {
			this.ttl = ttl;
		}
	}
	
	protected void logFailure(String message, Throwable e) {
		log.log(Level.SEVERE, message, e);
		FacesContext.getCurrentInstance().addMessage(null, new FacesMessage(message));
	}

	/**
	 * Parse ByteBuffer to String.
	 * 
	 * Just a stub for now.
	 * 
	 * @param buf ByteBuffer to parse
	 * @return String representation of buffer contents
	 */
	protected String parseBuffer(ByteBuffer buf) {
		try {
			Charset charset = Charsets.UTF_8;
			return charset.newDecoder().decode(buf.duplicate()).toString();
		} catch (CharacterCodingException e) {
			log.log(Level.SEVERE, "", e); //this should not happen in fact
			return "";
		}
	}
	
}
